/*  
  -- The Lime II Project -- 

  A tuplespaces-based middleware for coordinating agents and mobile hosts.
  Copyright (C) 2005.
  Gian Pietro Picco, Amy L. Murphy, Lorenzo Bellini.

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
*/

package lime2;
import java.util.*;

/**
 * Provides support for reconciliation of tuplespaces. Reconciliation is performed on a per host, per tuplespace
 * basis: we guarantee that tuplespaces that make use of a conservative policy are reconciled before any standard
 * operation takes place.
 * <br>
 * Reconciliation is not triggered by the engagement of a host (<i>eager</i> policy): it is triggered by the very
 * first communication which is sent or received from a remote host with whom we have been disengaged.
 * <br>
 * This <i>lazy</i> policy would weaken the model too much: misplaced tuples would be delivered only when a server
 * needs to communicate with its peer. So we force reconciliation also to run when two hosts stay connected for 
 * "enough" time (reconcile <i>all</i>: per host, for all the shared tuplespaces).
 * <br>
 * We guarantee that all the (class/instance) synchronized methods of Reconciliator terminate in a "small" amount of
 * time.
 * 
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini
 */

public class Reconciliator 
{
 // this hashtable[LimeServerID,hashtable[name,Reconciliator]] tracks Reconciliator instances per host, per tuplespace
 private static Hashtable newHosts; 
 
 // this hashtable[LimeServerID,hashtable[name,Long]] tracks how much time an host stays connected (.. since this 
 // information is used to enable "non lazy" reconciliation it must be per tuplespace - String)
 private static Hashtable noLazy;

 // this hashtable[name,Object] is a facility for synchronization when the LimeTSMgr doesn't exist (.. REJECT case)
 private static Hashtable resync;
 
 // this hashtable[thread-name,Boolean] keeps track of recursive reconciliations
 private static Hashtable recursive;
 
 // private non-static data members
 private LimeServerID target;
 private String tuplespace;
 private boolean reconciled; 
 private boolean reconciling;
 
 // static initializer
 static 
 {
  newHosts = new Hashtable();
  noLazy = new Hashtable();
  resync = new Hashtable();
  recursive = new Hashtable();
  new NoLazyReconciler(noLazy).start(); 
 }
 
 // for custom reconciliation. Added by Xing Jiankuan @ 2008-8-3
 static ArrayList reconJobs = new ArrayList();
 
 public static void AddReconciliationJob(ReconciliationJob job) {
	 synchronized (reconJobs) {
		 reconJobs.add(job);
	 }
 }
 
 public static void RemoveReconciliationJob(ReconciliationJob job) {
	 synchronized (reconJobs) {
		 reconJobs.remove(job);
	 }
 }
 
 /**
  * Constructor method (private). 
  * 
  * @param target host
  * @param tuplespace to reconcile
  */
 private Reconciliator(LimeServerID target, String tuplespace)
 {
  this.target = target;
  this.tuplespace = tuplespace;
  reconciled = false;
  reconciling = false;
 }
  
 /**
  * A new host has joined the system (.. we keep track of this for reconciliation). Synchronized (on class).
  * 
  * @param host "arriving" host
  */
 synchronized static void welcome(LimeServerID host)
 {
  newHosts.put(host, new Hashtable());	
  // code that enables "non lazy" reconciliation  
  Hashtable spaces = (Hashtable) noLazy.get(host);
  if (spaces == null) { spaces = new Hashtable(); noLazy.put(host, spaces); }
  String[] mynames = LimeTupleSpace.getNames();
  Long now = new Long(new Date().getTime());
  for (int i = 0; i < mynames.length; i++) spaces.put(mynames[i], now);  
 }
 
 /**
  * An old host has leaved the system (.. we keep track of this for reconciliation). Synchronized (on class).
  * 
  * @param host "departing" host
  */
 synchronized static void away(LimeServerID host)
 {
  newHosts.remove(host); 	
  // code that enables "non lazy" reconciliation
  noLazy.remove(host);
 }

 /**
  * Resets all reconciliation information. Synchronized (on class).
  */
 synchronized static void clear()
 {
  newHosts.clear();	
  // code that enables "non lazy" reconciliation
  noLazy.clear();
 }
 
 /**
  * Returns the Reconciliator instance for the pair host, tuplespace. Synchronized (on class).
  * 
  * @param target host
  * @param tuplespace name 
  * @return specific Reconciliator instance
  */
 synchronized static Reconciliator get(LimeServerID target, String tuplespace)
 {
  Reconciliator ret = null;	
  Hashtable spaces = (Hashtable) newHosts.get(target);	
  
  if (spaces == null) { spaces = new Hashtable(); newHosts.put(target, spaces); } 
  ret = (Reconciliator) spaces.get(tuplespace);
  if (ret == null) { ret = new Reconciliator(target, tuplespace); spaces.put(tuplespace, ret); }

  return ret; 
 }
  
 /**
  * Sets success/insuccess of reconciliation. Synchronized (on instance).
  * 
  * @param value success/insuccess
  */
 synchronized void setReconciled(boolean value)
 {
  reconciled = value;	
  // code that enables "non lazy" reconciliation
  Hashtable spaces = (Hashtable) noLazy.get(target);
  if (spaces == null) { spaces = new Hashtable(); noLazy.put(target, spaces); }
  if (value) {
	  spaces.put(tuplespace, new Long(-1)); // -1 is the conventional timer value for reconciled hosts 
	  //do post custom reconciliation jobs
	  for(int i = 0; i < reconJobs.size(); i++) {
		  ((ReconciliationJob)reconJobs.get(i)).postReconciliaton(tuplespace, target);
	  }
  }
  else // .. for passive reconciliation - the host considers itself reconcilied, it receives a reconciliation request
  	   // and the process fails: the passive side will have to re-try later, too 
  {
   String[] myspaces = LimeTupleSpace.getNames();
   boolean isAlsoMine = false;
   for (int i = 0; !isAlsoMine && i < myspaces.length; i++) if (tuplespace.equals(myspaces[i])) isAlsoMine = true;
   if (isAlsoMine) // we pay attention to the REJECT case: the requester failed whereas the replier succeded
   {	
    Long past = (Long) spaces.get(tuplespace);
    if (past != null && past.longValue() == -1) spaces.put(tuplespace, new Long(new Date().getTime())); 
   }
  }
 }

 /**
  * Sets reconciliation in progress/not in progress.
  * 
  * @param value success/insuccess
  */
 void setReconciling(boolean value)
 {
  synchronized (recursive) { recursive.put(Thread.currentThread().getName(), new Boolean(value)); }	
  reconciling = value;	
 }
 
 /**
  * Queries reconcilied status.
  * 
  * @return <code>true</code> or <code>false</code>
  */
 boolean isReconciled()
 {
  return reconciled;	
 }
 
 /**
  * Queries reconciling status.
  * 
  * @return <code>true</code> or <code>false</code>
  */
 boolean isReconciling()
 {
  return reconciling;	
 }
 
 /**
  * Returns whether the current thread is performing a reconciliation within another one (.. because out of misplaced
  * / ghost tuples or installation of reactions can "activate" a reaction listener.. and the reaction listener can 
  * start a "nested" reconciliation with another / the same host. Reaction listeners are executed by the Lime system
  * - which would be ran by the same thread of the first reconciliation process).
  *  
  * @return <code>true</code> or <code>false</code>
  */
 static boolean isNestedReconciliation()
 {
  boolean retrecur = false;	
  synchronized (recursive)
  {
   String threadname = Thread.currentThread().getName();
   Boolean boorecur = (Boolean) recursive.get(threadname); 
   // .. initialization (very "first" reconciliation)
   if (boorecur == null) recursive.put(threadname, new Boolean(false)); else retrecur = boorecur.booleanValue(); 
  }
  return retrecur;
 }
 
 /**
  * Returns a per tuplespace object for synchronization (REJECT case). Maybe we want to clear the hashtable when it 
  * grows huge.
  *  
  * @param name tuplespace name
  */
 synchronized static Object resync(String name)
 {
  Object oret = resync.get(name);
  if (oret == null) { oret = new Object(); resync.put(name, oret); }
  return oret;
 }
 
}

/**
 * This component prevents reconciliation from being "lazy". If two hosts stay connected for <i>enough</i> time they 
 * will try to reconcile (all) their tuplespaces. Since this mechanism uses group information of the 
 * LimeSystemTupleSpace, turn on FORCE_MAX_REFRESH (see LimeConstants) for maximum consistency.
 * 
 * @author Gian Pietro Picco
 * @author Amy Murphy
 * @author Lorenzo Bellini
 */

class NoLazyReconciler extends Thread
{
 // private data member 
 private Hashtable noLazy;
 
 /**
  * Constructor method.
  * 
  * @param noLazy "noLazy" table of Reconcilator
  */
 NoLazyReconciler(Hashtable noLazy)
 {
  super();	
  this.noLazy = noLazy;	
 }
	
 /**
  * Run method.
  */
 public void run()
 {
  // you can choose among:
  // - reconcileAllOld
  // - reconcileOldest	
  reconcileOldest();	
 }
 
 /**
  * Reconciles all old tuplespaces:hosts.
  */	
 private void reconcileAllOld()
 {
  for (;;)
  {	
   // sleeps	 
   try { Thread.sleep(LimeConstants.NO_LAZY_SLEEP_TIME); } catch (InterruptedException ie) { ie.printStackTrace(); }
  
   // gets names of local tuplespaces
   String[] mynames = LimeTupleSpace.getNames();
     
   // for each host of the "perceived" group
   for (Enumeration e = noLazy.keys(); e.hasMoreElements();)
   {
    LimeServerID targetHost = (LimeServerID) e.nextElement();
    Hashtable spaces = (Hashtable) noLazy.get(targetHost);
    if (spaces != null)
    {	
     // for each local tuplespace	
     for (int i = 0; i < mynames.length; i++) 
  	 {
      // the tuplespace has been recently created on this host: maybe it has been recently created on the peer, too
      // .. we have to try to reconciliate it
  	  if (spaces.get(mynames[i]) == null) spaces.put(mynames[i], new Long(new Date().getTime()));
  	  // extracts timer information
  	  Long past = (Long) spaces.get(mynames[i]);
      if (past != null) 
      {
       long xpast = past.longValue();
       // if (1) not reconciled and (2) perceived connected for "enough" time
       if ((xpast != -1) && (new Date().getTime() >= xpast + LimeConstants.NO_LAZY_TIME))
       	 doReconciliation(mynames[i], targetHost);
      } 
     } 	
    }	
   }	
  }
 }
 
 /**
  * Reconciles oldest tuplespace:host. 
  */	
 private void reconcileOldest()
 {
  for (;;)
  {	
   // auxiliary variables	
   LimeServerID qqqTarget = null;	
   String qqqName = null;	
   long qqqTime = Long.MAX_VALUE;
   
   // sleeps	 
   try { Thread.sleep(LimeConstants.NO_LAZY_SLEEP_TIME); } catch (InterruptedException ie) { ie.printStackTrace(); }
  
   // gets names of local tuplespaces
   String[] mynames = LimeTupleSpace.getNames();
     
   // for each host of the "perceived" group
   for (Enumeration e = noLazy.keys(); e.hasMoreElements();)
   {
    LimeServerID targetHost = (LimeServerID) e.nextElement();
    Hashtable spaces = (Hashtable) noLazy.get(targetHost);
    if (spaces != null)
    {	
     // for each local tuplespace	
     for (int i = 0; i < mynames.length; i++) 
  	 {
      // the tuplespace has been recently created on this host: maybe it has been recently created on the peer, too
      // .. we have to try to reconciliate it
  	  if (spaces.get(mynames[i]) == null) spaces.put(mynames[i], new Long(new Date().getTime()));
  	  // extracts timer information
  	  Long past = (Long) spaces.get(mynames[i]);
      if (past != null) 
      {
       long xpast = past.longValue();
       // if (1) not reconciled and (2) perceived connected for "enough" time
       if ((xpast != -1) && (new Date().getTime() >= xpast + LimeConstants.NO_LAZY_TIME) && (xpast < qqqTime))
       {
        qqqTarget = targetHost;
        qqqName = mynames[i];
        qqqTime = xpast;
       }
      } 
     } 	
    }	
   }
   
   if (qqqTarget != null && qqqName != null)
	   doReconciliation(qqqName, qqqTarget);
  }
 } 
 
 /**
  * Perform the reconciliation jobs. User can add and remove custom jobs with
  * {@link Reconciliator#AddReconciliationJob(ReconciliationJob)} and {@link 
  * Reconciliator#RemoveReconciliationJob(ReconciliationJob)}.
  * @param tsName
  * @param host
  * @author Xing Jiankuan
  */
 private void doReconciliation(String tsName, LimeServerID host) {
	 	//do the conventional reconciliation jobs
		CommunicationMgr.get().postEvent(
				new RemReconcileOp(LimeConstants.RECONCILE_NO_LAZY,
						LimeConstants.OPEN, tsName, host));
		
		//do the custom reconciliation jobs
		Iterator it = Reconciliator.reconJobs.iterator();
		while(it.hasNext()) {
			((ReconciliationJob)it.next()).doReconciliation(tsName, host);
		}
	}
}